package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.pojo.*;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.service.LoadPoint0F37Service;
import com.navinfo.tripanalysis.offline.service.impl.AbstractLoadPoint0F37Service;
import com.navinfo.tripanalysis.offline.util.PointUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.sql.Row;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.JavaConversions;

import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * 一汽通过Hive加载0F37协议数据
 * @author hmc
 */
@RunningType(value = RunningTypeEnum.HS, interfaceClass =LoadPoint0F37Service.class )
public class LoadPoint0F37ServiceHiveImpl extends AbstractLoadPoint0F37Service {

    private static final Logger logger = LoggerFactory.getLogger(LoadPoint0F37ServiceHiveImpl.class);
    /**
     * 覆盖默认的TID字段
     */
    private String TID_NAME = "terminalid";

    /**
     * 初始化加载字段列表
     */
    public LoadPoint0F37ServiceHiveImpl() {
        COL_NAMES.addAll(Arrays.asList(TID_NAME, "gpstime", "latitude", "longitude", "height",
                "speed","staticsdatamap","originallat", "originallng"));
    }

    /**
     * 获取hive的where条件
     * @param param
     * @return
     */
    @Override
    protected String getWhereSql(LoadDataParam param) {
        //分批处理
        String batchWhere = PointUtils.getBatchWhereSql(param);

        List<Long> tidList = param.getTidList();
        StringBuilder sb = new StringBuilder(StringUtils.isEmpty(batchWhere) ? "" : batchWhere+" and ")
            .append(" gpstime != 0 and gpstime is not null and "+TID_NAME+" != 0 and "+TID_NAME+" is not null ")
            //加载特定tid
            .append(CollectionUtils.isNotEmpty(tidList) ?
                    String.format(" and %s in(%s) ", TID_NAME, CommonUtils.mkString(tidList, ",")) : "")
            //协议字段
            .append(isSupportProtocolType() ?
                    " and ("+PROTOCOL_NAME+" is null or "+ PROTOCOL_NAME +"=0 ) " : "");

        return sb.toString();
    }

    /**
     * 获取HDFS的where条件
     * @param param
     * @return
     */
    @Override
    protected String getHDFSWhereSql(LoadDataParam param){
        List<Long> tidList = param.getTidList();

        StringBuilder sb = new StringBuilder(" gpstime != 0 and gpstime is not null and "+TID_NAME+" != 0 and "+TID_NAME+" is not null ")
                //加载特定tid
                .append(CollectionUtils.isNotEmpty(tidList) ?  String.format(" and %s in(%s) ", TID_NAME, CommonUtils.mkString(tidList, ",")) : "")
                //协议字段
                .append(isSupportProtocolType() ? " and ("+PROTOCOL_NAME+" is null or "+ PROTOCOL_NAME +"=0 ) " : "");

        return sb.toString();
    }

    @Override
    protected Point getPoint(Row row) {
        Point0F37 point = new Point0F37();
        point.setTid(Long.parseLong(Optional.ofNullable(row.getAs("terminalid")).orElse("0").toString()));
        point.setProtocol(ProtocolEnum.REALTIME);
        point.setGpsTime(Long.parseLong(Optional.ofNullable(row.getAs("gpstime")).orElse("0").toString())*1000L);
        point.setOriginalLat(Integer.parseInt(Optional.ofNullable(row.getAs("originallat")).orElse("0").toString()));
        point.setOriginalLng(Integer.parseInt(Optional.ofNullable(row.getAs("originallng")).orElse("0").toString()));
        point.setLatitude(Integer.parseInt(Optional.ofNullable(row.getAs("latitude")).orElse("0").toString()));
        point.setLongitude(Integer.parseInt(Optional.ofNullable(row.getAs("longitude")).orElse("0").toString()));
        point.setHeight(Double.parseDouble(Optional.ofNullable(row.getAs("height")).orElse("0").toString()));
        //point.setEngineOutputTorque(((int)row.getAs("engineoutputtorque"))/ 100);
        point.setSpeedKmh(Double.parseDouble(Optional.ofNullable(row.getAs("speed")).orElse("0").toString()));
        //处理附加信息
        Object statiscdataObj = Optional.ofNullable(row.getAs("staticsdatamap")).orElse(null);
        if(statiscdataObj != null){
            Map<String,String> statiscdatamap = JavaConversions.mapAsJavaMap(
                    (scala.collection.immutable.Map<String,String>)statiscdataObj);
            setStaticsdataMap(point,statiscdatamap);
        }
        //无大气温度属性
        //point.setAtmosphericTem(((int)row.getAs("atmospherictemperature"))/ 100);

        return point;
    }

    private static void setStaticsdataMap (Point0F37 point , Map<String,String> staticsdataMap){
        if(staticsdataMap != null && !staticsdataMap.isEmpty()){
            for(Map.Entry<String,String> map : staticsdataMap.entrySet()){
                String key = String.valueOf(map.getKey());
                String value = com.navinfo.tripanalysis.common.util.PointUtils.getField(Optional.ofNullable(map.getValue()).orElse("0"));
                if(value != "FF" && !"FF".equals(value)) {
                    switch (key) {
                        //仪表车速
                        case "1624":
                            point.setSpeedKmh(Double.parseDouble(value));
                            break;
                        //转速
                        case "190":
                            point.setRotation((int) (Double.parseDouble(value)));
                            break;
                        //发动机扭矩
                        case "513":
                            point.setEngineOutputTorque((int) (Double.parseDouble(value)));
                            break;
                        //离合器开关,0：松开，1：踩下
                        case "598":
                            point.setClutchSwitch((int) (Double.parseDouble(value)));
                            break;
                        //油门开度
                        case "91":
                            point.setAccPedalPos((int) (Double.parseDouble(value)));
                            break;
                        //发动机瞬时油耗
                        case "184":
                            point.setRealTimeOilConsumption(Double.parseDouble(value));
                            //发动机燃油消耗率
                        case "183":
                            point.setFuelConsumptionRate(Double.parseDouble(value));
                            break;
                        //制动脚踏开关,0：松开，1：踩下
                        case "597":
                            point.setBrake((int) (Double.parseDouble(value)));
                            break;
                        default:
                            break;
                    }
                }
            }
        }
    }
}
